package com.syncomponents.client.jms;

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;

import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jms.connection.ConnectionFactoryUtils;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.support.JmsUtils;
import org.springframework.util.Assert;

import com.syncomponents.client.MessageHelper;
import commonj.sdo.DataObject;
import commonj.sdo.helper.XMLDocument;
import commonj.sdo.helper.XMLHelper;

/**
 * jms客户端
 * 
 * @author sunkey
 * @date Mar 4, 2014
 */
public class JmsClient extends JmsTemplate {

    private static final transient Logger LOG = LoggerFactory.getLogger(JmsClient.class);

    // TODO 暂时没实现
    private int maxRetry = 3;
    
    private Destination receiveDestination;

    public JmsClient(ConnectionFactory connectionFactory) {
        super(connectionFactory);
    }

    public JmsClient() {
        super();
    }

    public void send(Object msg, String queueName) throws JMSException {
        doSend(queueName, msg, false, null);
    }

    public void send(Object msg) throws JMSException {
        doSend(getDefaultDestinationName(), msg, false, null);
    }

    /**
     * 
     * @param msg
     * @param returnType only 支持String ,DataObject
     * @return
     * @throws JMSException
     */
    public <T> T send(Object msg, Class<T> returnType) throws JMSException {
        return doSend(getDefaultDestinationName(), msg, true, returnType);
    }

    /**
     * 
     * @param queueName
     * @param msg
     * @param returnType only 支持String ,DataObject
     * @return
     * @throws JMSException
     */
    public <T> T send(String queueName, final Object msg, Class<T> returnType) throws JMSException {
        return doSend(queueName, msg, true, returnType);
    }

    protected <T> T doSend(String queueName, final Object msg, boolean isSyn, Class<T> returnType) throws JMSException {
        if (LOG.isDebugEnabled()) {
            LOG.debug("isSyn:{}, request:{}, isSyn:{}, msg:{}", new Object[] { isSyn, msg, isSyn, msg });
        }

        Connection conToClose = null;
        Session sessionToClose = null;
        Destination replyTo = null;
        try {
            conToClose = createConnection();
            sessionToClose = createSession(conToClose);
            conToClose.start();

            if (isSyn) {
                replyTo = sessionToClose.createTemporaryQueue();
            } else {
                replyTo = receiveDestination;
            }

            MessageCreator messageCreator = createMessageCreator(msg, isSyn, replyTo);
            Destination destination = resolveDestinationName(sessionToClose, queueName);
            Map<String, String> propertyMap = resolveActionName(msg);
            doSend(sessionToClose, destination, messageCreator, propertyMap);

            if (isSyn) {
                return doReceive(sessionToClose, returnType, replyTo);
            }
            return null;
        } finally {
            try {
                if (null != conToClose) {
                    JmsUtils.closeSession(sessionToClose);
                }
                if (null != sessionToClose) {
                    ConnectionFactoryUtils.releaseConnection(conToClose, getConnectionFactory(), false);
                }
            } catch (Exception e) {
            }
        }

    }

    protected Map<String, String> resolveActionName(final Object msg) {
        Map<String, String> propertyMap = new HashMap<String, String>();
        try {
            XMLDocument doc = XMLHelper.INSTANCE.load(msg.toString());
            DataObject dobj = doc.getRootObject();
            propertyMap.put("msgType", dobj.getString("msgType"));
            propertyMap.put("action", dobj.getString("action"));
            return propertyMap;
        } catch (Exception e) {
            return null;
        }

    }

    /**
     * Send the given JMS message.
     * 
     * @param session the JMS Session to operate on
     * @param destination the JMS Destination to send to
     * @param messageCreator callback to create a JMS Message
     * @throws JMSException if thrown by JMS API methods
     */
    protected void doSend(Session session, Destination destination, MessageCreator messageCreator,
            Map<String, String> propertyMap) throws JMSException {

        Assert.notNull(messageCreator, "MessageCreator must not be null");
        MessageProducer producer = createProducer(session, destination);
        try {
            Message message = messageCreator.createMessage(session);
            if (null != propertyMap) {
                message.setStringProperty("action", propertyMap.get("action"));
                message.setJMSCorrelationID(propertyMap.get("msgType"));
            }
            if (logger.isDebugEnabled()) {
                logger.debug("Sending created message: " + message);
            }
            doSend(producer, message);
            // Check commit - avoid commit call within a JTA transaction.
            if (session.getTransacted() && isSessionLocallyTransacted(session)) {
                // Transacted session created by this template -> commit.
                JmsUtils.commitIfNecessary(session);
            }
        } finally {
            JmsUtils.closeMessageProducer(producer);
        }
    }

    protected MessageCreator createMessageCreator(final Object msg, final boolean isSyn, final Destination replyTo) {
        return new MessageCreator() {
            @Override
            public Message createMessage(Session session) throws JMSException {
                if (msg instanceof String) {
                    logger.info("sending TextMessage");
                    TextMessage textMessage = session.createTextMessage(msg.toString());
                    textMessage.setJMSReplyTo(replyTo);
                    return textMessage;
                } else {
                    logger.info("sending  objectMessage");
                    ObjectMessage objectMessage = session.createObjectMessage();
                    objectMessage.setObject((Serializable) msg);
                    objectMessage.setJMSReplyTo(replyTo);
                    return objectMessage;
                }
            }
        };
    }

    @SuppressWarnings("unchecked")
    protected <T> T doReceive(Session session, Class<T> returnType, Destination replyTo) throws JMSException {

        Message message = doReceive(session, replyTo, null);
        
        if(message == null){
            new JMSException("return null, please check exception from server");
        }

        if (message instanceof TextMessage) {
            TextMessage textMessage = (TextMessage) message;
            try {
                if (returnType.equals(DataObject.class)) {
                    return (T) MessageHelper.toDataObject(textMessage);
                } else {
                    return (T) MessageHelper.toString(textMessage);
                }
            } catch (ClassNotFoundException e) {
                LOG.error(e.getMessage(), e);
                throw new JMSException(e.getMessage());
            }
        } else {
            ObjectMessage objectMessage = (ObjectMessage) message;
            return (T) objectMessage.getObject();
        }
    }

    public int getMaxRetry() {
        return maxRetry;
    }

    public void setMaxRetry(int maxRetry) {
        this.maxRetry = maxRetry;
    }

    public Destination getReceiveDestination() {
        return receiveDestination;
    }

    public void setReceiveDestination(Destination receiveDestination) {
        this.receiveDestination = receiveDestination;
    }

}
